package com.example.mq1.mqserver.core;

import com.example.mq1.common.Consumer;
import com.example.mq1.common.ConsumerEnv;
import com.example.mq1.common.MqException;
import com.example.mq1.mqserver.VirtualHost;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;

/**
 * 消费者控制中心
 */
public class ConsumerManager {

    private VirtualHost parent;
    private ExecutorService pool = Executors.newFixedThreadPool(4);
    //用来存放队列的名字
    private BlockingQueue<String> tokenQueue = new LinkedBlockingQueue<>();
    //扫描线程
    private Thread sannerThread;

    public ConsumerManager(VirtualHost virtualHost) {
        this.parent = virtualHost;

        //创建一个扫描线程取扫描队列，取出消息并消费
        sannerThread = new Thread(() -> {
            try {
                while(true) {
                    String queueName = tokenQueue.take();
                    MSGQueue queue = parent.getMemoryDataCenter().getQueue(queueName);
                    if(queue == null) {
                        throw new MqException("[ConsumerManager] 队列不存在！queueName=" + queueName);
                    }
                    //消费消息
                    synchronized (queue) {
                        consumeMessage(queue);
                    }
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        });

        sannerThread.setDaemon(true);
        sannerThread.start();
    }

    /**
     * 唤醒消费者(向 blockingQueue 中添加队列名 => 唤醒扫描线程阻塞等待)
     * @param queueName
     * @throws InterruptedException
     */
    public void notifyConsumer(String queueName) throws InterruptedException {
        tokenQueue.put(queueName);
    }

    /**
     * 添加消费者
     * @param consumerTag
     */
    public void addConsumer(String consumerTag, String queueName, boolean autoAck, Consumer consumer) {
        //1.虚拟主机隔离
        try {
            //2.找到对应队列
            MSGQueue queue = parent.getMemoryDataCenter().getQueue(queueName);
            if(queue == null) {
                throw new MqException("[ConsumerManager] 找不到对应的队列！queueName=" + queueName);
            }
            //3.创建出消费者
            ConsumerEnv consumerEnv = new ConsumerEnv(consumerTag, queueName, autoAck, consumer);
            synchronized (queue) {
                //4.将消费者放入到对应的队列 List 中
                queue.addConsumerEnv(consumerEnv);
                //5.如果当前队列中已经存在一些消息了，需要立即消费掉
                int n = parent.getMemoryDataCenter().getMessageCount(queueName);
                for(int i = 0; i < n; i++) {
                    //这个方法调用一次就消费一次消息
                    consumeMessage(queue);
                }
            }
        } catch (Exception e) {
            System.out.println("[ConsumerManager] 添加消费者失败！ consumerTag=" + consumerTag +
                    ", queueName=" + queueName);
            e.printStackTrace();
        }
    }

    /**
     * 消费指定队列的一个消息(消息确认(自动应答))
     * @param queue
     */
    private void consumeMessage(MSGQueue queue) {
        //1.按照轮询的方式先从队列中取出一个消费者
        ConsumerEnv consumerEnv = queue.chooseConsumerEnv();
        if(consumerEnv == null) {
            //这个队列没有消费者，暂时不用消费，等后面有了再消费
            return;
        }
        //2.获取该队列的一个消息
        Message message = parent.getMemoryDataCenter().pollMessage(queue.getQueueName());
        if(message == null) {
            //当前队列中没有消息了，不用消费
            return;
        }
        //3.使用线程池来执行回调任务
        pool.submit(() -> {
            try {
                // 1) 添加到未确认的消息中(这一步一定要在执行回调之前，这一步是为了保证消息即使在执行回调的过程中出错了，也不会导致消息丢失)
                parent.getMemoryDataCenter().addMessageWaitAck(queue.getQueueName(), message);
                // 2) 处理回调
                consumerEnv.getConsumer().handlerDelivery(consumerEnv.getConsumerTag(), message.getBasicProperties(), message.getBody());
                //3.内存/硬盘(deliverMode = 2) 删除消息
                if(message.getDeliverMode() == 2) {
                    parent.getDiskDataCenter().deleteMessage(queue, message);
                }
                parent.getMemoryDataCenter().removeMessageWaitAck(queue.getQueueName(), message.getMessageId());
                parent.getMemoryDataCenter().removeMessage(message.getMessageId());
            } catch (Exception e) {
                e.printStackTrace();
            }
        });
    }

}
